package com.patsnap.data.npd.dw.etl.base;

import com.patsnap.one.etl.cdc.CdcRecord;
import org.apache.flink.api.common.eventtime.Watermark;
import org.apache.flink.api.common.eventtime.WatermarkGenerator;
import org.apache.flink.api.common.eventtime.WatermarkOutput;

public class RateLimitWatermarkGenerator implements WatermarkGenerator<CdcRecord> {
    private final long maxRecordsPerSecond;
    private final long intervalMillis;
    private final long watermarkDelayMillis;
    private long counter;
    private long lastEmitTime;

    public RateLimitWatermarkGenerator(long maxRecordsPerSecond, long watermarkDelayMillis) {
        this.maxRecordsPerSecond = maxRecordsPerSecond;
        this.intervalMillis = 1000L / maxRecordsPerSecond;
        this.watermarkDelayMillis = watermarkDelayMillis;
        this.counter = 0;
        this.lastEmitTime = 0;
    }

    @Override
    public void onEvent(CdcRecord event, long eventTimestamp, WatermarkOutput output) {
        long currentTimestamp = System.currentTimeMillis();
        long elapsedMillis = currentTimestamp - lastEmitTime;

        // 根据时间间隔限制处理速度
        if (elapsedMillis < intervalMillis) {
            try {
                Thread.sleep(intervalMillis - elapsedMillis);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }

        counter++;
        lastEmitTime = System.currentTimeMillis();
        output.emitWatermark(new Watermark(eventTimestamp - watermarkDelayMillis));
    }

    @Override
    public void onPeriodicEmit(WatermarkOutput output) {
        // 不需要额外处理
    }
}